package com.atguigu.day09;

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternFlatSelectFunction;
import org.apache.flink.cep.PatternFlatTimeoutFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.List;
import java.util.Map;

//CEP迟到事件处理  自定义keyedProcessFunction实现
public class Example2 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env.addSource(new SourceFunction<OrderEvent>() {
            @Override
            public void run(SourceContext<OrderEvent> ctx) throws Exception {
                OrderEvent o1 = new OrderEvent("order-1", "create", 1000L);
                ctx.collectWithTimestamp(o1, o1.timestamp);
                Thread.sleep(1000L);
                OrderEvent o2 = new OrderEvent("order-2", "create", 2000L);
                ctx.collectWithTimestamp(o2, o2.timestamp);
                Thread.sleep(1000L);
                OrderEvent o3 = new OrderEvent("order-1", "pay", 3000L);
                ctx.collectWithTimestamp(o3, o3.timestamp);
                Thread.sleep(1000L);
                ctx.emitWatermark(new Watermark(7000L));
                Thread.sleep(1000L);
                OrderEvent o4 = new OrderEvent("order-2", "pay", 8000L);
                ctx.collectWithTimestamp(o4, o4.timestamp);
            }

            @Override
            public void cancel() {

            }
        })
                .keyBy(r -> r.orderId)
                .process(new KeyedProcessFunction<String, OrderEvent, String>() {
                    private ValueState<OrderEvent> orderState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        orderState = getRuntimeContext().getState(new ValueStateDescriptor<OrderEvent>("order-state", OrderEvent.class));
                    }

                    @Override
                    public void processElement(OrderEvent value, Context ctx, Collector<String> out) throws Exception {
                        if (value.type.equals("create")) {
                            orderState.update(value);
                            ctx.timerService().registerEventTimeTimer(value.timestamp + 5000L);
                        } else if (value.type.equals("pay") && orderState.value() != null) {
                            out.collect("订单: " + value.orderId + "支付成功");
                            orderState.clear();
                        }
                    }

                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
                        super.onTimer(timestamp, ctx, out);
                        if (orderState.value() != null && orderState.value().type.equals("create")) {
                            out.collect("订单： " + orderState.value().orderId + "未支付");
                            orderState.clear();
                        }
                    }
                })
                .print();

        env.execute();
    }

    public static class OrderEvent {
        public String orderId;
        public String type;
        public Long timestamp;

        @Override
        public String toString() {
            return "OrderEvent{" +
                    "orderId='" + orderId + '\'' +
                    ", type='" + type + '\'' +
                    ", timestamp=" + timestamp +
                    '}';
        }

        public OrderEvent() {
        }

        public OrderEvent(String orderId, String type, Long timestamp) {
            this.orderId = orderId;
            this.type = type;
            this.timestamp = timestamp;
        }
    }
}
